package eventBusV5

import (
	"errors"
	"sync"
	"time"
)

const (
	numWorkers       = 32    // 增加worker数量
	taskQueueSize    = 65536 // 增加队列缓冲大小
	maxRetryAttempts = 3     // 最大重试次数
)

type Bus struct {
	subMu       sync.RWMutex     // 保护subNode
	subNode     map[string]*node // topic -> node
	taskQueue   chan asyncTask   // 异步任务队列
	shutdownSig chan struct{}    // 关闭信号
	closed      bool             // 关闭标记
	closeMu     sync.Mutex       // 保护closed标志
}

type asyncTask struct {
	subs []Sub
	msg  interface{}
}

type node struct {
	subs []Sub
	mu   sync.RWMutex // 细粒度锁保护subs
}

type Sub struct {
	out chan interface{}
}

func NewBus() *Bus {
	b := &Bus{
		subNode:     make(map[string]*node),
		taskQueue:   make(chan asyncTask, taskQueueSize),
		shutdownSig: make(chan struct{}),
	}
	for i := 0; i < numWorkers; i++ {
		go b.worker()
	}
	return b
}

func (b *Bus) Subscribe(topic string, sub Sub) {
	b.subMu.Lock()
	defer b.subMu.Unlock()

	n, ok := b.subNode[topic]
	if !ok {
		n = &node{}
		b.subNode[topic] = n
	}

	n.mu.Lock()
	defer n.mu.Unlock()
	n.subs = append(n.subs, sub)
}

func (b *Bus) UnSubscribe(topic string, sub Sub) {
	b.subMu.Lock()
	defer b.subMu.Unlock()

	n, ok := b.subNode[topic]
	if !ok {
		return
	}

	n.mu.Lock()
	defer n.mu.Unlock()
	for i := range n.subs {
		if n.subs[i] == sub {
			n.subs = append(n.subs[:i], n.subs[i+1:]...)
			break
		}
	}

	if len(n.subs) == 0 {
		delete(b.subNode, topic)
	}
}

func (b *Bus) Publish(topic string, msg interface{}) error {
	b.subMu.RLock()
	n, ok := b.subNode[topic]
	b.subMu.RUnlock()

	if !ok {
		return errors.New("topic not exist")
	}

	n.mu.RLock()
	subs := make([]Sub, len(n.subs))
	copy(subs, n.subs)
	n.mu.RUnlock()

	task := asyncTask{subs: subs, msg: msg}

	for attempts := 0; attempts < maxRetryAttempts; attempts++ {
		select {
		case b.taskQueue <- task:
			return nil
		default:
			// 队列满时处理策略：打印告警并清空队列
			b.clearTaskQueue()
		}
	}
	return errors.New("task queue is full after max retry attempts")
}

func (b *Bus) PublishWithTimeout(topic string, msg interface{}, timeout time.Duration) error {
	b.subMu.RLock()
	n, ok := b.subNode[topic]
	b.subMu.RUnlock()

	if !ok {
		return errors.New("topic not exist")
	}

	n.mu.RLock()
	subs := make([]Sub, len(n.subs))
	copy(subs, n.subs)
	n.mu.RUnlock()

	task := asyncTask{subs: subs, msg: msg}

	for attempts := 0; attempts < maxRetryAttempts; attempts++ {
		select {
		case b.taskQueue <- task:
			return nil
		case <-time.After(timeout):
			// 队列满时处理策略：打印告警并清空队列
			b.clearTaskQueue()
		}
	}
	return errors.New("task queue is full after max retry attempts")
}

func (b *Bus) clearTaskQueue() {
	for len(b.taskQueue) > 0 {
		<-b.taskQueue
	}
}

func (b *Bus) worker() {
	for {
		select {
		case task, ok := <-b.taskQueue:
			if !ok {
				return
			}
			for _, sub := range task.subs {
				select {
				case sub.out <- task.msg:
				default:
					// 非阻塞发送，避免worker阻塞
					go func(sub Sub, msg interface{}) {
						sub.out <- msg
					}(sub, task.msg)
				}
			}
		case <-b.shutdownSig:
			return
		}
	}
}

func (b *Bus) Close() {
	b.closeMu.Lock()
	defer b.closeMu.Unlock()
	if b.closed {
		return
	}
	b.closed = true

	close(b.shutdownSig)
	close(b.taskQueue)

	b.subMu.Lock()
	defer b.subMu.Unlock()
	for _, n := range b.subNode {
		n.mu.Lock()
		for _, sub := range n.subs {
			close(sub.out)
		}
		n.subs = nil
		n.mu.Unlock()
	}
	b.subNode = nil
}

func NewSub() Sub {
	return Sub{out: make(chan interface{}, 1024)}
}

func (s *Sub) Out() <-chan interface{} {
	return s.out
}
